package SparkExercises.kejian3;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;

import java.util.Arrays;
import java.util.List;

public class SparkExercises {
    /**
     * 给定两份数据: （本市违章记录）和（临市B违章记录），
     * 请写一个spark程序查找车辆AK0803在本市以及临市B的交通违章记录。
     * 说明：两份数据结构相同（日期 设备编号 车牌号 违章类型码），
     * 列与列之间，均已tab作为分隔符。
     *
     * 思路：
     * 1）用textFile分别读取两个文件，得到两个RDD对象
     * 2）filter过滤出包含AK0803的元素
     * 3）求交集 [AK0803, X02]
     */
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("Exercises").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> rdd1 = sc.textFile("./data/records.txt");
        JavaRDD<String> rdd2 = sc.textFile("./data/recordsCityB.txt");
//        JavaRDD<String> rdd3 = rdd1.union(rdd2);

        JavaRDD<String> rdd4 = rdd1.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                int result = s.indexOf("AK0803");
                if (result != -1) {
                    return true;
                } else {
                    return false;
                }
            }
        });
        JavaRDD<String> rdd5 = rdd2.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String s) throws Exception {
                int result = s.indexOf("AK0803");
                if (result != -1) {
                    return true;
                } else {
                    return false;
                }
            }
        });

        JavaRDD<List> rdd6 = rdd4.map(new Function<String, List>() {
            @Override
            public List<String> call(String s) throws Exception {
                return Arrays.asList(s.split("\t")[2], s.split("\t")[3]);
            }
        });
        JavaRDD<List> rdd7 = rdd5.map(new Function<String, List>() {
            @Override
            public List<String> call(String s) throws Exception {
                return Arrays.asList(s.split("\t")[2], s.split("\t")[3]);
            }
        });
        JavaRDD<List> intersection = rdd6.intersection(rdd7);
        List<List> res1 = intersection.collect();

//        List<List> res = rdd6.collect();
        for (List i : res1){
            System.out.println(i);
        }

    }
}
